package com.liyong.res.service;

import com.liyong.res.util.KafkaConstant;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.PartitionInfo;

import java.time.Duration;
import java.util.*;

/**
 * create by yongli on 2020-05-17 22:17
 */

public class ConsumerTest {

    public void consumer() {

        Properties props = new Properties();

        props.put("bootstrap.servers", KafkaConstant.KAFKA_SERVER_ADDRESS);
        props.put("group.id", "1");
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put("session.timeout.ms", "30000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        Consumer<String, String> consumer = new KafkaConsumer<>(props);
        // 订阅主题
        consumer.subscribe(Arrays.asList(KafkaConstant.KAFKA_TOPIC_NAME));
        consumer.seekToBeginning(new ArrayList<>());

        // ===== 拿到所有的topic ===== //
        Map<String, List<PartitionInfo>> listTopics = consumer.listTopics();
        Set<Map.Entry<String, List<PartitionInfo>>> entries = listTopics.entrySet();

//        for (Map.Entry<String, List<PartitionInfo>> entry : entries) {
//            System.out.println("the key is :" + entry.getKey() + ",the value is:" + entry.getValue());
//        }

        // 死循环
        Thread thread = new Thread(() -> {
            while (true) {
                // 时刻去kafka轮训有没有消息
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(10));
                for (ConsumerRecord<String, String> record : records) {
                    // 可以看到partion的内容,sout最原始的方法
                    System.out.println("[fetched from partition " + record.partition() + ", offset: " +
                            record.offset() + ", message: " + record.value() + "]");
                }
            }
        });
        thread.start();

    }


}
